import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.connector.jdbc.JdbcInputFormat;

public class StreamTableSync {
    public static void main(String[] args) throws Exception {
        // 1. 初始化流处理执行环境（关键：使用StreamExecutionEnvironment）
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 从A表读取数据 (MySQL示例)
        String sourceQuery = "SELECT id, name, age FROM student";
        // 修复：使用DataStream接收，而非DataSet
        DataStream<Tuple3<Integer, String, Integer>> sourceData1 = env.createInput(
                JdbcInputFormat.buildJdbcInputFormat()
                        .setDrivername("com.mysql.cj.jdbc.Driver")
                        .setDBUrl("jdbc:mysql://host.docker.internal:3306/ryvue?useSSL=false&serverTimezone=UTC")
                        .setUsername("root")
                        .setPassword("password")
                        .setQuery(sourceQuery)
                        .setRowTypeInfo(new RowTypeInfo(
                                Types.INT,    // id
                                Types.STRING, // name
                                Types.INT     // age
                        ))
                        .finish()
        ).map(new MapFunction<Row, Tuple3<Integer, String, Integer>>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple3<Integer, String, Integer> map(Row row) {
                return new Tuple3<>(
                        row.getFieldAs(0),
                        row.getFieldAs(1),
                        row.getFieldAs(2)
                );
            }
        });

        // 3. 转换数据
//        DataStream<Tuple3<Integer, String, Integer>> transformedData = sourceData1
//                .map(tuple -> new Tuple3<>(tuple.f0, tuple.f1, tuple.f2));

        // 4. 写入B表（修复：使用JdbcSink和Builder模式）
        sourceData1.addSink(JdbcSink.sink(
                "INSERT INTO student2 (id, name, age) VALUES ( ?, ?)",
                (statement, tuple) -> {
                    statement.setString(1, tuple.f1);    // 设置第二个参数（name）
                    statement.setInt(2, tuple.f2);       // 设置第三个参数（age）
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(1000)                // 批量写入大小
                        .withBatchIntervalMs(200)           // 批量写入间隔（毫秒）
                        .withMaxRetries(3)                  // 失败重试次数
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://host.docker.internal:3306/ryvue?useSSL=false&serverTimezone=UTC")
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUsername("root")
                        .withPassword("password")
                        .build()
        ));

        // 5. 执行任务
        env.execute("test.StreamTableSync Table A to Table B");
    }
}
